package com.atguigu.flink.chapter07;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/9/17 9:24
 */
public class Flink13_Sideout_2 {
    
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
    
        DataStreamSource<Integer> stream = env.fromElements(10, 20, 11, 1, 3, 68, 10, 13);
    
        SingleOutputStreamOperator<Integer> oddStream = stream
            .process(new ProcessFunction<Integer, Integer>() {
                @Override
                public void processElement(Integer value,
                                           Context ctx,
                                           Collector<Integer> out) throws Exception {
                    if (value % 2 == 1) {
                        out.collect(value);
                    }else{
                        ctx.output(new OutputTag<Integer>("even"){}, value);
                    }
                }
            });
    
        DataStream<Integer> evenStream = oddStream.getSideOutput(new OutputTag<Integer>("even") {});
    
        oddStream.print("奇数");
        evenStream.print("偶数");
    
        env.execute();
        
    }
}
/*
1. 存储迟到数据
2. 分流

 */